/*
 * @Author: zdc 3215400595@qq.com
 * @Date: 2023-08-06 23:42:57
 * @LastEditors: zdc 3215400595@qq.com
 * @LastEditTime: 2023-08-08 15:47:36
 * @FilePath: /Logs_System/looper.hpp
 * @Description: 这是默认设置,请设置`customMade`, 打开koroFileHeader查看配置 进行设置: https://github.com/OBKoro1/koro1FileHeader/wiki/%E9%85%8D%E7%BD%AE
 */
// 实现异步工作器

#ifndef _M_LOOP_H__
#define _M_LOOP_H__
#include "buffer.hpp"
#include <thread>
#include <mutex>
#include <condition_variable>
#include <functional>
#include <memory>
#include <atomic>
namespace tong_log
{
    using Functor = std::function<void(Buffer &)>;

    enum class AsyncType
    {
        ASYNC_SAFE,  // 安全状态，就阻塞，避免资源浪费
        ASYNC_UNSAFE // 不考虑资源耗尽的问题，有数据就添加，无限扩容，用于性能测试
    };
    class AsyncLooper
    {
    public:
        using ptr = std::shared_ptr<AsyncLooper>;
        AsyncLooper(const Functor &callback, AsyncType looper_type = AsyncType::ASYNC_SAFE)
            : _callBack(callback),
              _looper_type(AsyncType::ASYNC_SAFE),
              _stop(false),
              _thread(std::thread(&AsyncLooper::threadEntry, this)) {}
        ~AsyncLooper()
        {
            stop();
        }

    public:
        void stop()
        {
            _stop = true;
            _cond_con.notify_all(); // 唤醒所有的工作线程
            _thread.join();         // 等待工作线程的退出
        };
        void push(const char *data, size_t len) // 添加数据的接口
        {
            std::unique_lock<std::mutex> lock(_mutex);
            // 条件变量控制，如果缓冲区剩余空间大于数据长度，则可以添加数据
            if (_looper_type == AsyncType::ASYNC_SAFE)
                _cond_pro.wait(lock, [&]()
                               { return _pro_buf.writeAbleSize() >= len; });
            // 能走下来，表示满足条件，向缓冲区中添加数据
            _pro_buf.push(data, len);
            _cond_con.notify_one(); // 唤醒一个消费者对缓冲区的数据处理
        }

    private:
        void threadEntry() // 线程入口函数
        {
            while (1)
            {
                // 不对数据的处理过程加锁
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    
                    // 退出标志位被设置，并且生产缓冲区没有数据，这时候再退出，否则就会造成生产缓冲区中有数据，但是没有处理完毕
                    if (_stop && _pro_buf.empty())
                        break;

                    // 当前退出前，被唤醒，或者有数据被唤醒，返回真，继续向下进行，否则重新休眠
                    _cond_con.wait(lock, [&]()
                                   { return _stop || !_pro_buf.empty(); });

                    // 生产缓存区有内容，交换缓冲区
                    _con_buf.swap(_pro_buf);
                    // 唤醒生产者
                    if (_looper_type == AsyncType::ASYNC_SAFE)
                        _cond_pro.notify_all();
                }
                // 消费者被唤醒后，对消费缓冲区进行数据处理
                _callBack(_con_buf);
                // 初始化缓冲区
                _con_buf.reset();
            }
        }

    private:
        Functor _callBack; // 回调函数，具体缓存区数据进行处理的回调函数，有异步工作器使用者传入
    private:
        std::atomic<bool> _stop; // 工作器停止标志，是否停止异步工作器，当条件变量不满足的时候
        Buffer _pro_buf;         // 生产缓冲区
        Buffer _con_buf;         // 消费缓冲区
        std::mutex _mutex;
        std::condition_variable _cond_pro;
        std::condition_variable _cond_con;
        std::thread _thread; // 异步工作器，对应的工作线程

        AsyncType _looper_type;
    };

}
#endif